/**
 * Date:2021年10月13日 上午9:48:32
 * Copyright (c) 2021, www.bwbroad.com All Rights Reserved
 */

package com.xue.redis;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamInfo.XInfoGroups;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

/**
 * Description: Redis Stream 是 Redis 5.0 版本新增加的数据结构 <br/>
 * Date:2021年10月13日 上午9:48:32 <br/>
 * 参考 https://blog.csdn.net/hhl18730252820/article/details/114826366
 * https://www.zhihu.com/question/279540635
 * 
 * @author xuejianxin
 */
@Configuration
public class RedisStreamConfig implements ApplicationListener<ApplicationStartedEvent> {
	public static final String TOPIC_MYSTREAM = "mystream";
	public static final String GROUP_MYSTREAM_1 = "mystreamGroup1";
	public static final String GROUP_MYSTREAM_2 = "mystreamGroup2";
	// @Autowired
	// private MystreamListener mystreamListener;

	// 订阅流,如果需要订阅多个流,复制这个方法就可以
	// @Bean
//	public Subscription subscription1(final RedisTemplate<String, String> redisTemplate) throws UnknownHostException {
//		// 订阅的选项
//		var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
//				.pollTimeout(Duration.ofSeconds(1)).build();
//		var listenerContainer = StreamMessageListenerContainer.create(redisTemplate.getConnectionFactory(), options);
//		// 手动添加信息到 mystream
//		// XADD mystream * field1 value1 field2 value2 field3 value3
//		// 手动添加 消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer),需要手动添加mystreamGroup1
//		// 使用xgroup命令创建了一个mystreamGroup1消费者组，该消费者组与mystream进行了关联，以后mystreamGroup1消费者组中的消费者就会mystream中拿数据；
//		// 符号"$"代表mystream中目前最大的ID，消费者拿到的entry的id一定会大于此刻$代表的最大ID。你也可以指定这个最大的ID，比如0
//		// XGROUP CREATE mystream mystreamGroup1 $
//		// 消息ID消息ID的形式是timestampInMillis-sequence，例如1527846880572-5，它表示当前的消息在毫米时间戳1527846880572时产生，并且是该毫秒内产生的第5条消息。
//		// 消息ID可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加入的消息的ID要大于前面的消息ID
//		// XGROUP CREATE mystream mystreamGroup1 0-0
//		// XINFO GROUPS mystream
//
//		XInfoGroups groupInfo = redisTemplate.opsForStream().groups(TOPIC_MYSTREAM);
//		// 判断消费组是否存在
//		boolean groupIsExist = groupInfo.stream().anyMatch(g -> g.groupName().equalsIgnoreCase(GROUP_MYSTREAM_1));
//		// 如果不存在则创建
//		if (!groupIsExist) {
//			System.out.printf("%s %s not exist,create it! \r\nXGROUP CREATE %s %s $ \r\n", TOPIC_MYSTREAM,
//					GROUP_MYSTREAM_1, TOPIC_MYSTREAM, GROUP_MYSTREAM_1);
//			redisTemplate.opsForStream().createGroup(TOPIC_MYSTREAM, GROUP_MYSTREAM_1);
//		}
//
//		// 创建监听
//		StreamListener<String, MapRecord<String, String, String>> mystreamListener = (
//				MapRecord<String, String, String> entries) -> {
//			System.out.println("message id " + entries.getId());
//			System.out.println("stream " + entries.getStream());
//			System.out.println("body " + entries.getValue());
//			try {
//				TimeUnit.SECONDS.sleep(2);// 休息2s
//			} catch (InterruptedException e) {
//				e.printStackTrace();
//			}
//		};
//
//		// 自动ack
//		var subscription = listenerContainer.receiveAutoAck(
//				// 这里把当前hostname 添加到消费组mystreamGroup 中
//				// 每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动
//				Consumer.from(GROUP_MYSTREAM_1, InetAddress.getLocalHost().getHostName()),
//				StreamOffset.create(TOPIC_MYSTREAM, ReadOffset.lastConsumed()), mystreamListener);
//		// 开始监听
//		listenerContainer.start();
//		return subscription;
//	}

	// 手动回复
	@Bean
	public StreamMessageListenerContainer<String, MapRecord<String, String, String>> subscription2(
			final RedisTemplate<String, String> redisTemplate) throws UnknownHostException {
		// 订阅的选项
		var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
				// 一次性最多拉取多少条消息
				.batchSize(10) // 这个和pending 列表长度有关，一次性拉取10条，如果batch size为1的话不容易丢消息
				// 消息消费异常的handler
				.errorHandler(e -> {
					e.printStackTrace();
				})
				// 超时时间，设置为0，表示不超时（超时后会抛出异常）
				.pollTimeout(Duration.ofSeconds(10))
				// 序列化器
				// .serializer(new StringRedisSerializer())
				.build();
		var listenerContainer = StreamMessageListenerContainer.create(redisTemplate.getConnectionFactory(), options);

		// 判断流是否存在
		var keyIsExist = redisTemplate.execute((RedisConnection connection) -> {
			return connection.exists(TOPIC_MYSTREAM.getBytes());
		});

		if (!keyIsExist) {// 不存在则创建
			// XADD mystream * field1 value1 field2 value2 field3 value3
			redisTemplate.opsForStream().add(TOPIC_MYSTREAM, Map.of("start", "true"));
		}

		XInfoGroups groupInfo = redisTemplate.opsForStream().groups(TOPIC_MYSTREAM);
		// 判断消费组是否存在
		boolean groupIsExist = groupInfo.stream().anyMatch(g -> g.groupName().equalsIgnoreCase(GROUP_MYSTREAM_2));
		// 如果不存在则创建
		if (!groupIsExist) {
			System.out.printf("%s %s not exist,create it! \r\nXGROUP CREATE %s %s $ \r\n", TOPIC_MYSTREAM,
					GROUP_MYSTREAM_2, TOPIC_MYSTREAM, GROUP_MYSTREAM_2);
			redisTemplate.opsForStream().createGroup(TOPIC_MYSTREAM, GROUP_MYSTREAM_2);
		}
		// 创建监听，这个类可以从外部注入，此处为了修改方便 写成了本地类
		StreamListener<String, MapRecord<String, String, String>> mystreamListener = (
				MapRecord<String, String, String> entries) -> {
			RecordId recordId = entries.getId();
			System.out.println("####message id " + recordId);
			System.out.println("####stream " + entries.getStream());
			System.out.println("####body " + entries.getValue());
			// 保存偏移id
			redisTemplate.opsForValue().set(TOPIC_MYSTREAM + "_OFFSET", recordId.toString());
			// *手动ack,最好保存偏移id
			redisTemplate.opsForStream().acknowledge(TOPIC_MYSTREAM, GROUP_MYSTREAM_2, recordId);

			try {
				TimeUnit.SECONDS.sleep(2);// 休息2s
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

		};

		String consumerId = InetAddress.getLocalHost().getHostName();
		// ReadOffset.lastConsumed() 在batchSize>1时会丢失消息
		// StreamOffset<String> streamOffset = StreamOffset.create(TOPIC_MYSTREAM,
		// ReadOffset.lastConsumed());
		//用0-0 会阻塞不读取不知道为什么
		// StreamOffset<String> streamOffset = StreamOffset.create(TOPIC_MYSTREAM,
		// ReadOffset.from("0-0"));
		// 获取偏移id
		//用 xinfo groups mystream 查看pendling 列表
		ReadOffset offSet = Optional.ofNullable(redisTemplate.opsForValue().get(TOPIC_MYSTREAM + "_OFFSET"))
				.map(of -> ReadOffset.from(of)).orElse(ReadOffset.lastConsumed());
		StreamOffset<String> streamOffset = StreamOffset.create(TOPIC_MYSTREAM, offSet);
		Consumer consumer = Consumer.from(GROUP_MYSTREAM_2, consumerId);
		// https://segmentfault.com/q/1010000038547927
		// 连接中途有过断网的情况，但即使redis已经正常链接了，messageListener也无法再次收到任何关于通道的消息，
		// 这让人很郁闷，不能保证中途一次网都不会断，但一旦断网，redis stream的消息就失效了，
		// 解决这个问题需要在 在运行时抛出异常，不取消任务执行
		StreamMessageListenerContainer.StreamReadRequest<String> readRequest = StreamMessageListenerContainer.StreamReadRequest
				.builder(streamOffset).cancelOnError(throwable -> {
					// *此处非常关键, 查询超时，有可能时断网了，不能取消,在运行时抛出异常，不取消任务执行
					return !(throwable instanceof QueryTimeoutException);
				}).consumer(consumer).autoAcknowledge(false)// 不自动回复
				.build();
		listenerContainer.register(readRequest, mystreamListener);

		// 光接收不ack
		// PEL如何避免消息丢失
		// 在客户端消费者读取Stream消息时，Redis服务器将消息回复给客户端的过程中，客户端突然断开了连接，消息就丢失了。
		// 但是PEL里已经保存了发出去的消息ID。待客户端重新连上之后，可以再次收到PEL中的消息ID列表。
		// 不过此时xreadgroup的起始消息ID不能为参数>，
		// 而必须是任意有效的消息ID，一般将参数设为0-0，表示读取所有的PEL消息以及自last_delivered_id之后的新消息。
		/*
		 * var subscription = listenerContainer.receive( // 这里把当前hostname
		 * 添加到消费组mystreamGroup 中 // 每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标
		 * last_delivered_id 往前移动 // ReadOffset.from("0")
		 * 设置offset从0开始时导致启动后无法监听消息，有卡顿，原因待查!!! Consumer.from(GROUP_MYSTREAM_2,
		 * InetAddress.getLocalHost().getHostName()),
		 * StreamOffset.create(TOPIC_MYSTREAM,ReadOffset.from("0") ), mystreamListener);
		 * //这里设置为0 很重要不丢消息 ReadOffset.from("0")
		 */
		// 开始监听
		// listenerContainer.start(); //在 onApplicationEvent中实现
		// return subscription;
		return listenerContainer;
	}

	// 容器启动后启动redis stream 监听
	@Override
	public void onApplicationEvent(ApplicationStartedEvent event) {
		if (event.getApplicationContext().getParent() == null) {// root application context 没有parent，他就是老大.
			// 需要执行的逻辑代码，当spring容器初始化完成后就会执行该方法。
			event.getApplicationContext()
					// 相当于返回js中的闭包函数，函数式编程思想，主要式方便链式调用，在这里并不能延迟初始化
					.getBeanProvider(StreamMessageListenerContainer.class).ifAvailable(Lifecycle::start);
			System.out.println("-----------------------start---------");
		}

	}

}
